package com.ymj.netty.client;

import com.ymj.netty.common.ClientRequest;
import com.ymj.netty.common.Response;
import com.ymj.netty.common.ResponseCodeEnum;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;


/**
 *  @Description: 异步获取调用结果，此类的
 *  @author: yanmj
 *  @Date: 2020-06-07
 */
public class ClientFuture {

    /*存放所有的future对象*/
    private static ConcurrentHashMap<Long,ClientFuture> allClientFuture ;
    private Response response;
    private final Lock lock = new ReentrantLock();
    private Condition condition = lock.newCondition();
    private long timeout = 20*1000;  //超时时间，默认20s
    private long startTime = System.currentTimeMillis();

    static {
        allClientFuture = new ConcurrentHashMap<>();
        //启动定时任务，处理那些超时的任务
        ScheduledThreadPoolExecutor schedule = new ScheduledThreadPoolExecutor(1);
        schedule.scheduleAtFixedRate(() -> {
            try {
                ConcurrentHashMap.KeySetView<Long, ClientFuture> keySet = allClientFuture.keySet();
                keySet.forEach(k -> {
                    ClientFuture clientFuture = allClientFuture.get(k);
                    if (clientFuture == null) {
                        allClientFuture.remove(clientFuture);
                        //如果超时
                    } else if (clientFuture.getTimeout() < System.currentTimeMillis() - clientFuture.getStartTime()) {
                        Response response = new Response();
                        response.setId(k);
                        response.setCode(ResponseCodeEnum.TIMEOUT.getCode());
                        response.setMsg(ResponseCodeEnum.TIMEOUT.getMsg());
                        receive(response);
                    }
                });
            } catch (Exception e) {
                e.printStackTrace();
            }
        }, 1000, 3000, TimeUnit.MILLISECONDS);
    }

    /**
     * 每次构造实例的时候，都将自己注册到allClientHashMap
     * @param request
     */
    public ClientFuture(ClientRequest request) {
        allClientFuture.put(request.getId(),this);
    }

    /**
     * 当netty客户端接收到服务端发来的response，将其放到allClientFuture中，并唤醒阻塞等待结果的线程
     * 这也是异步获取结果的关键，netty客户端接收线程、等待结果的线程之间的异步通信
     * @param response
     */
    public static void receive(Response response){
        ClientFuture clientFuture = allClientFuture.get(response.getId());
        if(null != clientFuture){
            Lock lock = clientFuture.lock;
            lock.lock();
            try{
                clientFuture.setResponse(response);
                clientFuture.condition.signal(); //唤醒阻塞等待结果的线程
            }catch (Exception e){
                e.printStackTrace();
            }finally {
                lock.unlock();
            }
        }
    }


    /**
     * 调用远程方法的线程通过此方法获取调用结果
     * @return
     */
    public Response get() {
        lock.lock();
        try {
            while (!done()){
                this.condition.await(timeout, TimeUnit.MILLISECONDS);
                //如果超时
                if(System.currentTimeMillis()-startTime > timeout){
                    System.out.println("请求超时!");
                    break;
                }
            }
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            lock.unlock();
        }
        return this.response;
    }


    public Boolean done(){
        if(null != this.response){
            return true;
        }
        return false;
    }

    public void setResponse(Response response) {
        this.response = response;
    }

    public long getStartTime() {
        return startTime;
    }

    public long getTimeout() {
        return timeout;
    }

}
